use crate::core::blueprint::Blueprint;
use crate::core::error::blueprint_error::BlueprintError;
use crate::core::error::convert_error::{ConvertError, U32ToUSizeSnafu};
use crate::core::error::not_found_error::NotFoundError;
use crate::core::error::request_error::NotFoundSnafu;
use crate::core::error::ParrotError::InvalidOperation;
use crate::core::error::{request_error, ParrotError, RequestSnafu};
use grpc_server::State::Running;
use grpc_server::{BuildBlogParameter, Detail, Info, SyncLogRequest, SyncLogResponse, UpdateDocumentParameter, {
    CreateRequest, CreateResponse, FetchJobsRequest, FetchJobsResponse, PauseRequest,
    PauseResponse, ResumeRequest, ResumeResponse, SyncDetailRequest, SyncDetailResponse,
}};
use snafu::ResultExt;
use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, MutexGuard};
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::{Stream, StreamExt};
use tonic::{Request, Response, Status};
use grpc_server::create_request::JobType;
use crate::core::blueprint::build_blog::BuildBlog;
use crate::core::blueprint::build_coffee_tales::BuildAndroidCoffeeTales;
use crate::core::blueprint::update_document::UpdateDocument;
use crate::core::services::job::job::Job;

const PAGE_SIZE: usize = 10;

pub struct JobService {
    next_job_id: AtomicUsize,
    jobs: Mutex<Vec<Arc<Job>>>,
}

impl JobService {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn jobs(&self) -> MutexGuard<'_, Vec<Arc<Job>>> {
        self.jobs.lock().unwrap()
    }

    pub fn get_job(&self, id: usize) -> Result<Arc<Job>, NotFoundError> {
        self.jobs
            .lock()
            .unwrap()
            .iter()
            .find(|j| j.id() == id)
            .cloned()
            .ok_or(NotFoundError::Job { id })
    }

    pub fn run(&self, blueprint: Box<dyn Blueprint>) -> Result<Info, ParrotError> {
        let job = blueprint
            .build(self.next_job_id.load(Ordering::Relaxed))
            .map_err(Into::<BlueprintError>::into)?;
        let job = Arc::new(job);
        let mut jobs = self.jobs.lock().unwrap();
        (*jobs).push(job.clone());
        {
            let job = job.clone();
            tokio::spawn(async move { job.run().await });
        };

        self.next_job_id.fetch_add(1, Ordering::Relaxed);
        Ok(job.clone().info())
    }
}

impl Default for JobService {
    fn default() -> Self {
        Self {
            jobs: Mutex::new(vec![]),
            next_job_id: AtomicUsize::new(0),
        }
    }
}

#[tonic::async_trait]
impl grpc_server::job_service_server::JobService for JobService {
    async fn fetch_jobs(
        &self,
        request: Request<FetchJobsRequest>,
    ) -> Result<Response<FetchJobsResponse>, Status> {
        let page = request.get_ref().page;
        let page_usize: usize = page.try_into().unwrap();

        let infos: Vec<_> = (*self.jobs())
            .iter()
            .skip(page_usize * PAGE_SIZE)
            .take(PAGE_SIZE)
            .map(|j| j.info())
            .collect();

        Ok(Response::new(FetchJobsResponse { page, infos }))
    }

    async fn create(
        &self,
        request: Request<CreateRequest>,
    ) -> Result<Response<CreateResponse>, Status> {
        let job_type = request
            .into_inner()
            .job_type
            .ok_or(NotFoundError::Field {
                filed_name: "job_type".to_string(),
            })
            .context(request_error::NotFoundSnafu {})
            .context(RequestSnafu {})?;

        let blueprint: Box<dyn Blueprint> = match job_type {
            JobType::UpdateDocument(UpdateDocumentParameter { uri, dst }) => {
                let update_document = UpdateDocument::new(&uri, &dst).unwrap();
                Box::new(update_document)
            }
            JobType::BuildBlog(BuildBlogParameter { uri, dst }) => {
                let build_blog = BuildBlog::new(&uri, &dst).unwrap();
                Box::new(build_blog)
            }
            JobType::BuildAndroidCoffeeTales(p) => {
                let build_android_coffee_tales = BuildAndroidCoffeeTales::new(p).unwrap();
                Box::new(build_android_coffee_tales)
            }
        };
        let result = self
            .run(blueprint)
            .map(|i| Response::new(CreateResponse { info: Some(i) }))?;

        Ok(result)
    }

    async fn pause(
        &self,
        request: Request<PauseRequest>,
    ) -> Result<Response<PauseResponse>, Status> {
        Err(Status::unimplemented(""))
    }

    async fn resume(
        &self,
        _request: Request<ResumeRequest>,
    ) -> Result<Response<ResumeResponse>, Status> {
        Err(Status::unimplemented(""))
    }

    type SyncDetailStream = Pin<Box<dyn Stream<Item = Result<SyncDetailResponse, Status>> + Send>>;

    async fn sync_detail(
        &self,
        request: Request<SyncDetailRequest>,
    ) -> Result<Response<Self::SyncDetailStream>, Status> {
        let job_id = request.into_inner().id;
        let job_id_usize = job_id
            .try_into()
            .context(U32ToUSizeSnafu { value: job_id })
            .map_err(|e| <ConvertError as Into<ParrotError>>::into(e))?;

        let job = self
            .get_job(job_id_usize)
            .context(NotFoundSnafu {})
            .context(RequestSnafu {})?;

        if job.state() != Running {
            Err(InvalidOperation {
                message: "Cannot sync detail of a job which is not running".to_string(),
            })?
        }

        let change_receiver = job.context().change_receiver();
        let stream = BroadcastStream::new(change_receiver);
        let mapped = stream.map(move |s| {
            let info = job.info();
            let steps = job.steps().map(|s| s.pb_step()).collect();
            Ok(SyncDetailResponse {
                detail: Some(Detail {
                    info: Some(info),
                    steps,
                }),
            })
        });

        let mapped = Pin::new(Box::new(mapped));

        Ok(Response::new(mapped))
    }

    type SyncLogStream = Pin<Box<dyn Stream<Item = Result<SyncLogResponse, Status>> + Send>>;

    async fn sync_log(
        &self,
        request: Request<SyncLogRequest>,
    ) -> Result<Response<Self::SyncLogStream>, Status> {
        let job_id = request.into_inner().id;
        let job_id_usize = job_id
            .try_into()
            .context(U32ToUSizeSnafu { value: job_id })
            .map_err(|e| <ConvertError as Into<ParrotError>>::into(e))?;

        let job = self
            .get_job(job_id_usize)
            .context(NotFoundSnafu {})
            .context(RequestSnafu {})?;

        if job.state() != Running {
            Err(InvalidOperation {
                message: "Cannot sync detail of a job which is not running".to_string(),
            })?
        }

        let log_receiver = job.context().log_receiver();
        let stream = BroadcastStream::new(log_receiver);
        let mapped = stream.map(move |s| {
            Ok(SyncLogResponse {
                log: s.whatever_context::<_, ParrotError>("Failed to receive data from stream")?,
            })
        });

        let mapped = Pin::new(Box::new(mapped));

        Ok(Response::new(mapped))
    }
}
